package com.atguigu.gmall.realtime.util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.DriverManager;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/24 9:09
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> {
    
    private Connection conn;
    private ThreadPoolExecutor pool;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        String phoenixUrl = "jdbc:phoenix:hadoop162,hadoop163,hadoop164:2181";
        conn = DriverManager.getConnection(phoenixUrl);
        pool = MyThreadPoolUtil.getThreadPool();
    }
    
    @Override
    public void asyncInvoke(T input,
                            ResultFuture<T> resultFuture) throws Exception {
        pool.execute(new Runnable() {
            @Override
            public void run() {
                Jedis redisClient = MyRedisUtil.getRedisClient();
                // 和具体的业务相关
                addDim(input, redisClient); // 把维度添加到了input中
                resultFuture.complete(Collections.singletonList(input));
                redisClient.close();
            }
        });
    }
    
    public JSONObject readDim(Jedis redisClient, String tableName, Object id) {
        
        // 先从缓存读数据, 缓存没有, 再从HBase读数据
        String key = tableName + ":" + id;
        if (redisClient.exists(key)) {
            System.out.println(tableName + "的 id " + id + " 走缓存");
            String userInfoJson = redisClient.get(key);
            return JSON.parseObject(userInfoJson);
        } else {
            System.out.println(tableName + "的 id " + id + " 走数据库");
            String sql = "select * from " + tableName + " where ID = ?";
            JSONObject jsonObj = MyJDBCUtil
                .queryList(conn, sql, new Object[]{id}, JSONObject.class, false)
                .get(0);
            // 把数据存入缓存, 过期时间一周
            redisClient.setex(key, 60 * 60 * 24 * 7, jsonObj.toJSONString());
            return jsonObj;
        }
    }
    public abstract void addDim(T input, Jedis redisClient);
    
    @Override
    public void close() throws Exception {
        if (conn != null && !conn.isClosed()) {
            conn.close();
        }
        
        if (pool != null && !pool.isShutdown()) {
            pool.shutdown();
        }
    }
}
